package io.kimmking.dubbo.common.balance.impl;

import io.kimmking.spi.Join;

import java.util.List;

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.SortedMap;
import java.util.concurrent.ConcurrentSkipListMap;

@Join
public class HashLoadBalance extends AbstractLoadBalance {

    private static final int VIRTUAL_NODE_NUM = 5;

    /**
     *
     * @param upstreamList 服务器地址列表，形如:192.168.1.100:18080
     * @param from_ip 请求来源IP
     * @return
     */
    @Override
    protected String doSelect(final List<String> upstreamList, final String from_ip) throws Exception{
        final ConcurrentSkipListMap<Long, String> treeMap = new ConcurrentSkipListMap<>();
        for (String address : upstreamList) {
            for (int i = 0; i < VIRTUAL_NODE_NUM; i++) {
                long addressHash = hash("WEEK11-" + address + "-HASH-" + i);
                treeMap.put(addressHash, address);
            }
        }
        long hash = hash(String.valueOf(from_ip));
        SortedMap<Long, String> lastRing = treeMap.tailMap(hash);
        if (!lastRing.isEmpty()) {
            return lastRing.get(lastRing.firstKey());
        }
        return treeMap.firstEntry().getValue();
    }

    private static long hash(final String key) throws Exception {
        // md5 byte
        MessageDigest md5;
        try {
            md5 = MessageDigest.getInstance("MD5");
        } catch (Exception e) {
            throw new Exception("MD5 not supported", e);
        }
        md5.reset();
        byte[] keyBytes;
        keyBytes = key.getBytes(StandardCharsets.UTF_8);

        md5.update(keyBytes);
        byte[] digest = md5.digest();

        // hash code, Truncate to 32-bits
        long hashCode = (long) (digest[3] & 0xFF) << 24
                | ((long) (digest[2] & 0xFF) << 16)
                | ((long) (digest[1] & 0xFF) << 8)
                | (digest[0] & 0xFF);
        return hashCode & 0xffffffffL;
    }
}
